package com.wl.cloud.datax.common.utils;

import com.wl.cloud.datax.common.config.KafkaProperties;
import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.clients.admin.*;
import org.apache.kafka.common.KafkaFuture;
import org.apache.kafka.common.config.ConfigResource;
import org.apache.kafka.common.config.SaslConfigs;

import java.util.*;

/**
 * kafka客户端工具
 */
public class CustomKafkaAdminClientUtils {

    public static AdminClient adminClient(KafkaProperties kafkaProperties) {
        Properties properties = new Properties();
        properties.setProperty(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaProperties.getBootstrapServersConfig());
        properties.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, kafkaProperties.getSecurityProtocolConfig());
        properties.put(SaslConfigs.SASL_MECHANISM, kafkaProperties.getSaslMechanism());
        properties.put(SaslConfigs.SASL_JAAS_CONFIG, kafkaProperties.getSaslJaasConfig());
        return AdminClient.create(properties);
    }

    /**
     * 创建Topic实例
     */
    public static void createTopic(AdminClient adminClient, String topicName, int partitions, short replications) {
        NewTopic newTopic = new NewTopic(topicName, partitions, replications);
        CreateTopicsResult createTopicsResult = adminClient.createTopics(Collections.singletonList(newTopic));
        System.out.println("CreateTopicsResult : " + createTopicsResult);
    }

    /**
     * 获取Topic列表
     */
    public static void topicList(AdminClient adminClient) throws Exception {
        // 是否查看internal选项
        ListTopicsOptions options = new ListTopicsOptions();
        // 设置我们是否应该列出内部topic
        options.listInternal(true);

        // 列出集群中可用的topic
        ListTopicsResult listTopicsResult = adminClient.listTopics(options);
        // 返回一个topic名称集合的future(这里是KafkaFuture)
        Set<String> names = listTopicsResult.names().get();
        // 返回一个KafkaFuture，它产生一个 TopicListing 对象的集合
        Collection<TopicListing> topicListings = listTopicsResult.listings().get();
        // 返回一个KafkaFuture，它产生一个topic名称到 TopicListing 对象的映射。
        KafkaFuture<Map<String, TopicListing>> mapKafkaFuture = listTopicsResult.namesToListings();
        // 打印names
        names.forEach(System.out::println);
        System.out.println("---------------------------topic列表-------------------------");
        // 打印topicListings
        topicListings.forEach(System.out::println);
        System.out.println("---------------------------topic列表-------------------------");
    }

    /**
     * 删除topic
     */
    public static void delTopics(AdminClient adminClient, String topicName) throws Exception {
        // 删除一批topic。
        // 此操作不是事务性的，因此它可能对某些主题成功，而对另一些主题则失败。
        // DeleteTopicsResult返回成功后，所有代理可能需要几秒钟才能意识到主题已消失。 在此期间，
        // listTopics()和describeTopics(Collection)可能会继续返回有关已删除主题的信息。
        DeleteTopicsResult deleteTopicsResult = adminClient.deleteTopics(Arrays.asList(topicName));
        deleteTopicsResult.all().get();
    }

    /**
     * 描述topic
     * name: yibo_topic
     * desc: (name=yibo_topic,
     *      internal=false,
     *      partitions=
     *          (partition=0,
     *          leader=192.168.174.128:9092 (id: 0 rack: null),
     *          replicas=192.168.174.128:9092 (id: 0 rack: null),
     *          isr=192.168.174.128:9092 (id: 0 rack: null)),
     *          authorizedOperations=null)
     * @throws Exception
     */
    public static void describeTopic(AdminClient adminClient, String topicName) throws Exception {
        // 描述集群中的一些topic。
        DescribeTopicsResult describeTopicsResult = adminClient.describeTopics(Arrays.asList(topicName));

        Map<String, TopicDescription> stringTopicDescriptionMap = describeTopicsResult.all().get();

        Set<Map.Entry<String, TopicDescription>> entries = stringTopicDescriptionMap.entrySet();
        System.out.println("----------------------------topic信息-----------------------------");
        entries.forEach((entry) -> {
            System.out.println("name ：" + entry.getKey() + " , desc: " + entry.getValue());
        });
        System.out.println("----------------------------topic信息-----------------------------");
    }

    /**
     * 查询配置信息
     */
    public static void describeConfig(AdminClient adminClient, String topicName) throws Exception {

        ConfigResource configResource = new ConfigResource(ConfigResource.Type.TOPIC, topicName);
        // 获取指定资源的配置
        DescribeConfigsResult describeConfigsResult = adminClient.describeConfigs(Arrays.asList(configResource));
        Map<ConfigResource, Config> configResourceConfigMap = describeConfigsResult.all().get();
        System.out.println("----------------------------配置信息-----------------------------");
        configResourceConfigMap.entrySet().forEach((entry) -> {
            System.out.println("configResource : " + entry.getKey() + " , Config : " + entry.getValue());
        });
        System.out.println("----------------------------配置信息-----------------------------");
    }


    /**
     * 修改配置信息 老版API
     */
    public static void alterConfig() throws Exception {
//        AdminClient adminClient = adminClient();
//
//        Map<ConfigResource, Collection<AlterConfigOp>> configMaps = new HashMap<>();
//        // 具有配置的资源的类，需要提供type和名称 Type是他内部维护的枚举类，共有四种类型：BROKER_LOGGER((byte) 8), BROKER((byte) 4), TOPIC((byte) 2),
//        // UNKNOWN((byte) 0)
//        ConfigResource configResource = new ConfigResource(ConfigResource.Type.TOPIC, TOPIC_NAME);
//        // 包含名称、值和操作类型的更改配置条目的类 ,需要注入ConfigEntry，和操作类型，同样OpType是个枚举类
//        AlterConfigOp alterConfigOp =
//                new AlterConfigOp(new ConfigEntry("preallocate", "false"), AlterConfigOp.OpType.SET);
//        configMaps.put(configResource, Arrays.asList(alterConfigOp));
//        // 逐步更新指定资源的配置
//        AlterConfigsResult alterConfigsResult = adminClient.incrementalAlterConfigs(configMaps);
//        alterConfigsResult.all().get();
    }


    /**
     * 增加partitions数量
     */
    public static void incrPartitions(AdminClient adminClient, String topicName ,int partitions) throws Exception {
        Map<String, NewPartitions> partitionsMap = new HashMap<>();

        NewPartitions newPartitions = NewPartitions.increaseTo(partitions);
        partitionsMap.put(topicName, newPartitions);
        CreatePartitionsResult createPartitionsResult = adminClient.createPartitions(partitionsMap);
        createPartitionsResult.all().get();

    }
}
